use bytes::Bytes;
use std::time::Duration;
use std::{borrow::BorrowMut, sync::Arc};
use tokio::sync::mpsc::Receiver;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::sync::Mutex;
use webrtc::{
    api::{
        interceptor_registry::register_default_interceptors, media_engine::MediaEngine, APIBuilder,
    },
    data_channel::{
        data_channel_message::DataChannelMessage, data_channel_state::RTCDataChannelState,
        RTCDataChannel,
    },
    ice_transport::ice_candidate::{RTCIceCandidate, RTCIceCandidateInit},
    interceptor::registry::Registry,
    peer_connection::{
        peer_connection_state::RTCPeerConnectionState,
        sdp::session_description::RTCSessionDescription, RTCPeerConnection,
    },
};

use super::cache::remove_peer_connection;
use crate::data_channels::request::request_response;
use crate::data_channels::{handler::on_message, message::Message};

use super::get_webrtc_config;

pub async fn new_offer_connection(
    remote_peer_id: &String,
) -> anyhow::Result<(
    Arc<RTCPeerConnection>,
    UnboundedSender<Message>,
    Vec<String>,
)> {
    //创建连接和数据通道
    let (peer_connection, sender, candidates) =
        _new_webrtc_connection(remote_peer_id, true).await?;
    //设置角色
    let offer = peer_connection.create_offer(None).await?;
    //
    let mut gather_complete = peer_connection.gathering_complete_promise().await;
    //
    peer_connection.set_local_description(offer).await?;
    let _ = gather_complete.recv().await;

    let candidates: Vec<String> = candidates.lock().await.clone();
    tracing::debug!("gathering is done");
    Ok((peer_connection, sender, candidates))
}

pub async fn new_answer_connection(
    remote_peer_id: &String,
    remote_candidates: &Vec<String>,
    local_description: &String,
) -> anyhow::Result<(
    Arc<RTCPeerConnection>,
    UnboundedSender<Message>,
    Vec<String>,
)> {
    //
    let (peer_connection, sender, candidates) =
        _new_webrtc_connection(remote_peer_id, false).await?;

    //设置远程
    let _ = set_remote_description_by_string(
        peer_connection.clone(),
        local_description,
        remote_candidates,
    )
    .await?;
    //
    let answer = peer_connection.create_answer(None).await?;

    tracing::debug!("gathering_complete_promise");
    let mut gather_complete = peer_connection.gathering_complete_promise().await;
    //
    tracing::debug!("set_local_description");
    peer_connection.set_local_description(answer).await?;

    let _ = gather_complete.recv().await;

    tracing::debug!("gathering is done");

    let candidates: Vec<String> = candidates.lock().await.clone();

    Ok((peer_connection, sender, candidates))
}
//
async fn _new_webrtc_connection(
    remote_peer_id: &String,
    offer: bool,
) -> anyhow::Result<(
    Arc<RTCPeerConnection>,
    UnboundedSender<Message>,
    Arc<Mutex<Vec<String>>>,
)> {
    //获取配置
    let config = get_webrtc_config()?;
    //初始化
    let mut m = MediaEngine::default();
    m.register_default_codecs()?;
    let mut registry = Registry::new();
    registry = register_default_interceptors(registry, &mut m)?;

    let api = APIBuilder::new()
        .with_media_engine(m)
        .with_interceptor_registry(registry)
        .build();

    let peer_connection = Arc::new(api.new_peer_connection(config).await?);

    let (cannel_tx, mut cannel_rx) = tokio::sync::mpsc::channel::<()>(1);

    let sender = if offer {
        _create_webrtc_data_channel(remote_peer_id, peer_connection.clone(), cannel_rx).await?
    } else {
        //
        _on_webrtc_data_channel(remote_peer_id, peer_connection.clone(), cannel_rx).await?
    };

    let remote_peer_id2 = remote_peer_id.clone();
    peer_connection.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
        tracing::debug!("连接状态改变: {s}");
        if s == RTCPeerConnectionState::Failed {
            //需要关闭连接
            if let Err(err) = remove_peer_connection(&remote_peer_id2) {
                tracing::error!("移除缓存中的连接失败!{}", err);
            }
        } else if s == RTCPeerConnectionState::Disconnected {
            //需要关闭连接
            if let Err(err) = remove_peer_connection(&remote_peer_id2) {
                tracing::error!("移除缓存中的连接失败!{}", err);
            }
        }

        Box::pin(async {})
    }));
    let candidates: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(vec![]));

    let candidates1 = candidates.clone();
    peer_connection.on_ice_candidate(Box::new(move |candidate: Option<RTCIceCandidate>| {
        let candidates_1 = candidates1.clone();

        Box::pin(async move {
            if let Some(candidate) = candidate {
                if let Ok(candidate) = candidate.to_json() {
                    let json_str = serde_json::to_string(&candidate).unwrap();
                    let value = model_graph_common::utils::base64::encode(json_str.as_str());
                    candidates_1.lock().await.push(value);
                } else {
                }
            } else {
            }
        })
    }));

    Ok((peer_connection, sender, candidates))
}

async fn _create_webrtc_data_channel(
    remote_peer_id: &String,
    peer_connection: Arc<RTCPeerConnection>,
    mut cannel_rx: Receiver<()>,
) -> anyhow::Result<UnboundedSender<Message>> {
    // 创建数据通道
    let data_channel = peer_connection.create_data_channel("g01", None).await?;

    let (receiver_tx, receiver_rx) = tokio::sync::mpsc::unbounded_channel::<Message>();

    if let Err(err) = _add_data_channel_handler(
        data_channel,
        receiver_tx.clone(),
        Arc::new(Mutex::new(receiver_rx)),
        remote_peer_id,
    )
    .await
    {
        tracing::error!("创建数据通道失败: {err}");
        return Err(anyhow::anyhow!("{}", err));
    }

    Ok(receiver_tx)
}

async fn _on_webrtc_data_channel(
    remote_peer_id: &String,
    peer_connection: Arc<RTCPeerConnection>,
    cannel_rx: Receiver<()>,
) -> anyhow::Result<UnboundedSender<Message>> {
    let (receiver_tx, receiver_rx) = tokio::sync::mpsc::unbounded_channel::<Message>();

    // 创建数据通道
    let receiver_tx_1 = receiver_tx.clone();
    let remote_peer_id_1 = remote_peer_id.clone();

    let receiver_rx_arc = Arc::new(Mutex::new(receiver_rx));

    peer_connection.on_data_channel(Box::new(move |data_channel: Arc<RTCDataChannel>| {
        let d_label = data_channel.label().to_owned();

        let d1 = data_channel.clone();
        let receiver_tx_1 = receiver_tx_1.clone();
        let remote_peer_id_1 = remote_peer_id_1.clone();
        let receiver_rx_arc = receiver_rx_arc.clone();

        tracing::debug!("创建数据通道.....");
        Box::pin(async move {
            if d_label != "g01" {
                tracing::error!("创建数据通道不是g01: {d_label}");
            } else {
                //
                if let Err(err) =
                    _add_data_channel_handler(d1, receiver_tx_1, receiver_rx_arc, &remote_peer_id_1)
                        .await
                {
                    tracing::error!("创建数据通道失败: {err}");
                }
            }
        })
    }));
    Ok(receiver_tx)
}

async fn _add_data_channel_handler(
    data_channel: Arc<RTCDataChannel>,
    receiver_tx: UnboundedSender<Message>,
    receiver_rx: Arc<Mutex<UnboundedReceiver<Message>>>,
    remote_peer_id: &String,
) -> anyhow::Result<()> {
    //读线程
    let d_label = data_channel.label().to_owned();
    let peer_id = remote_peer_id.clone();

    let sender = receiver_tx.clone();
    data_channel.on_message(Box::new(move |msg: DataChannelMessage| {
        let peer_id_1 = peer_id.clone();
        let sender_1 = sender.clone();

        Box::pin(async move {
            if msg.is_string {
                //字符串
                tracing::error!("消息写入的是字符串!");
            } else {
                //二进制
                if let Ok(message) = Message::try_from(msg.data.to_vec()) {
                    tracing::debug!("Message:{:?}", message);
                    match on_message(&peer_id_1, message).await {
                        Ok(response) => {
                            //查看是否需要返回
                            if let Some(res_msg) = response {
                                sender_1.send(res_msg);
                            }
                        }
                        Err(err) => {
                            tracing::error!("消息处理失败!{}", err);
                        }
                    }
                } else {
                    tracing::error!("消息反序列化失败!");
                }
            }
        })
    }));

    data_channel.on_close(Box::new(|| {
        tracing::info!("通道关闭!");
        Box::pin(async {})
    }));

    let d2 = Arc::clone(&data_channel);
    let remote_peer_id2 = remote_peer_id.clone();

    data_channel.on_open(Box::new(|| {
        tracing::info!("数据通道打开!",);

        //PING
        Box::pin(async move {
            // let mut result = anyhow::Result::<usize>::Ok(1);
            // while result.is_ok() {
            //     let timeout = tokio::time::sleep(Duration::from_secs(5));
            //     tokio::pin!(timeout);

            //     tokio::select! {
            //         _ = timeout.as_mut() =>{
            //             let message = Message::ping();
            //             tracing::debug!("Sending '{message}'");
            //             if let Ok(d) = TryInto::<Vec<u8>>::try_into(message){
            //                 result = d2.send(&Bytes::from(d)).await.map_err(|err|anyhow::anyhow!("{}",err));
            //             }
            //         }
            //     };
            // }
        })
    }));

    let d1 = Arc::clone(&data_channel);
    data_channel.on_error(Box::new(move |err| {
        tracing::error!("数据通道发生异常!{}", err);
        match d1.ready_state() {
            RTCDataChannelState::Closing => {
                tracing::debug!("数据通道正在关闭!");
            }
            RTCDataChannelState::Closed => {
                tracing::debug!("数据通道已经关闭!");
            }
            _ => {}
        }
        Box::pin(async {})
    }));

    let d1 = Arc::clone(&data_channel);
    //写线程
    let receiver_rx = receiver_rx.clone();

    tokio::spawn(async move {
        let d2 = Arc::clone(&d1);

        let receiver_rx = receiver_rx.clone();
        loop {
            let receiver_rx_1 = receiver_rx.clone();

            let mut receiver_rx_2 = {
                let r = receiver_rx_1.lock().await;
                r
            };
            tokio::select! {
                // _=cannel_rx.recv()=>{
                //     //退出
                //     tracing::error!("收到退出指令!");
                //     break;
                // }
                message = receiver_rx_2.recv() => {
                    if let Some(message) = message {
                        //
                        if let Ok(content) = TryInto::<Vec<u8>>::try_into(message){
                            let result  =  d2.send(&Bytes::from(content)).await;
                            if let Err(err) = result {
                                 tracing::error!("消息发送失败: {err}");
                                 break;
                             }else{
                                 tracing::debug!("消息发送成功!");
                             }
                        }else{
                            tracing::error!("消息序列化失败!");
                            continue;
                        }
                    }
                    //
                    match d2.ready_state(){
                        RTCDataChannelState::Connecting =>{

                        }
                        RTCDataChannelState::Open =>{

                        }
                        RTCDataChannelState::Closing =>{
                            tracing::debug!("数据通道正在关闭!");
                            break;
                        }
                        RTCDataChannelState::Closed =>{
                            tracing::debug!("数据通道已经关闭!");
                            break;
                        }
                        _=>{}
                    }
                }
            }
        }
        //
    });

    Ok(())
}

pub async fn set_remote_description_by_string(
    peer_connection: Arc<RTCPeerConnection>,
    local_description: &String,
    remote_candidates: &Vec<String>,
) -> anyhow::Result<()> {
    //设置远程
    let desc_data = model_graph_common::utils::base64::decode(local_description.as_str())?;
    if let Ok(c) = serde_json::from_str::<RTCSessionDescription>(&desc_data) {
        peer_connection.set_remote_description(c).await?;
    } else {
        tracing::error!("unmarshal_candidate error!");
    }

    for candidate in remote_candidates {
        let desc_data = model_graph_common::utils::base64::decode(candidate.as_str())?;
        if let Ok(candidate) = serde_json::from_str::<RTCIceCandidateInit>(&desc_data) {
            // peer_connection.set_remote_description(c).await?;
            if let Err(err) = peer_connection.add_ice_candidate(candidate).await {
                tracing::error!("{}", err);
            }
        } else {
            tracing::error!("unmarshal_candidate error!");
            break;
        }
    }
    Ok(())
}

pub async fn get_local_description_string(
    peer_connection: Arc<RTCPeerConnection>,
) -> anyhow::Result<String> {
    //
    if let Some(local_desc) = peer_connection.local_description().await {
        let json_str = serde_json::to_string(&local_desc)?;
        tracing::debug!("{json_str}");
        let local_description = model_graph_common::utils::base64::encode(&json_str);
        tracing::debug!("{local_description}");
        Ok(local_description)
    } else {
        tracing::error!("generate local_description failed!");
        return Err(anyhow::anyhow!("generate local_description failed!"));
    }
}
